/*!
 * \file ParserUDP.cpp
 * \project	WonderTrader
 *
 * \author Wesley
 * \date 2020/03/30
 * 
 * \brief UDP行情解析器实现文件
 * 
 * \details 该文件实现了ParserUDP类的所有功能，主要包括：
 *          - UDP网络协议的完整封装和管理
 *          - 基于boost::asio的异步网络编程实现
 *          - 双通道UDP数据接收（广播+服务器）
 *          - 多种行情数据类型的处理和分发
 *          - UDP数据包的结构化定义和解析
 *          - 高性能的网络数据缓冲和队列管理
 *          - 自动重连和网络异常恢复机制
 *          - 完整的日志记录和调试支持
 *          - 订阅请求的批量发送和管理
 *          - 线程安全的网络操作处理
 * 
 * \note 该实现专门针对UDP协议的高性能行情数据传输
 * \warning UDP协议不保证数据可靠性，实现中需要考虑数据丢失处理
 */
#include "ParserUDP.h"
#include "../Includes/WTSVariant.hpp"
#include "../Includes/WTSDataDef.hpp"

#include <boost/bind.hpp>

/*!
 * \brief 通用日志记录模板函数
 * \tparam Args 可变参数类型
 * \param sink 行情回调接口指针
 * \param ll 日志级别
 * \param format 格式字符串
 * \param args 可变参数列表
 * 
 * \details 提供格式化日志输出功能，支持：
 *          - 多种日志级别（INFO、ERROR、WARN、DEBUG）
 *          - 线程安全的格式化输出
 *          - 自动内存管理和缓冲区保护
 *          - 与WonderTrader日志系统集成
 * 
 * \note 使用fmtlib提供高性能格式化，线程本地存储确保安全性
 */
 //By Wesley @ 2022.01.05
#include "../Share/fmtlib.h"
template<typename... Args>
inline void write_log(IParserSpi* sink, WTSLogLevel ll, const char* format, const Args&... args)
{
	if (sink == NULL)
		return;

	static thread_local char buffer[512] = { 0 };
	memset(buffer, 0, 512);
	fmt::format_to(buffer, format, args...);

	sink->handleParserLog(ll, buffer);
}

/*!
 * \brief UDP消息类型定义
 * 
 * 定义UDP数据包中使用的各种消息类型常量
 */
#define UDP_MSG_SUBSCRIBE	0x100		///< 订阅请求消息类型
#define UDP_MSG_PUSHTICK	0x200		///< Tick行情推送消息类型
#define UDP_MSG_PUSHORDQUE	0x201		///< 委托队列推送消息类型
#define UDP_MSG_PUSHORDDTL	0x202		///< 委托明细推送消息类型
#define UDP_MSG_PUSHTRANS	0x203		///< 逐笔成交推送消息类型

#pragma pack(push,1)

/*!
 * \struct UDPPacketHead
 * \brief UDP数据包头结构
 * 
 * 所有UDP数据包的通用头部，包含消息类型信息
 */
typedef struct UDPPacketHead
{
	uint32_t		_type;		///< 消息类型（使用UDP_MSG_*常量）
} UDPPacketHead;

/*!
 * \struct _UDPReqPacket
 * \brief UDP请求数据包结构
 * 
 * 用于发送订阅请求等客户端到服务器的数据包
 */
typedef struct _UDPReqPacket : UDPPacketHead
{
	char			_data[1020];		///< 请求数据内容（最大1020字节）
} UDPReqPacket;

/*!
 * \struct UDPDataPacket
 * \brief UDP数据包模板结构
 * \tparam T 数据类型（如WTSTickStruct等）
 * 
 * 用于封装各种类型行情数据的通用UDP数据包结构
 */
template <typename T>
struct UDPDataPacket : UDPPacketHead
{
	T			_data;				///< 具体的数据内容
};
#pragma pack(pop)

/*!
 * \brief UDP数据包类型定义
 * 
 * 基于UDPDataPacket模板定义的具体数据包类型
 */
typedef UDPDataPacket<WTSTickStruct>	UDPTickPacket;		///< Tick行情数据包
typedef UDPDataPacket<WTSOrdQueStruct>	UDPOrdQuePacket;	///< 委托队列数据包
typedef UDPDataPacket<WTSOrdDtlStruct>	UDPOrdDtlPacket;	///< 委托明细数据包
typedef UDPDataPacket<WTSTransStruct>	UDPTransPacket;		///< 逐笔成交数据包

/*!
 * \brief C接口导出函数
 * 
 * 提供标准的动态库导出接口，支持插件化架构
 */
extern "C"
{
	/*!
	 * \brief 创建UDP行情解析器实例
	 * \return 新创建的行情解析器接口指针
	 * 
	 * 用于动态库插件系统，创建ParserUDP实例
	 */
	EXPORT_FLAG IParserApi* createParser()
	{
		ParserUDP* parser = new ParserUDP();
		return parser;
	}

	/*!
	 * \brief 销毁UDP行情解析器实例
	 * \param parser 要销毁的行情解析器接口指针（引用传递，销毁后置为NULL）
	 * 
	 * 安全地销毁ParserUDP实例并清理资源
	 */
	EXPORT_FLAG void deleteParser(IParserApi* &parser)
	{
		if (NULL != parser)
		{
			delete parser;
			parser = NULL;
		}
	}
};

/*!
 * \brief 构造函数
 * 
 * 初始化ParserUDP实例的各个成员变量为默认值
 */
ParserUDP::ParserUDP()
	: _b_socket(NULL)
	, _s_socket(NULL)
	, _strand(_io_service)
	, _stopped(false)
	, _sink(NULL)
	, _connecting(false)
{
}

/*!
 * \brief 析构函数
 * 
 * 清理ParserUDP实例，释放相关资源
 */
ParserUDP::~ParserUDP()
{
}

/*!
 * \brief 初始化UDP行情解析器
 * \param config 配置参数对象
 * \return 是否初始化成功
 * 
 * \details 从配置文件中读取UDP连接参数并初始化网络组件：
 *          - 服务器地址：host（UDP服务器主机地址）
 *          - 端口配置：bport（广播端口）、sport（服务器端口）
 *          - 性能参数：gpsize（分组大小，默认1000）
 *          - 网络地址解析：将主机地址解析为IP地址
 *          - 终端点配置：设置服务器终端点和广播终端点
 *          - 地址绑定：配置广播接收的本地绑定地址
 * 
 * \note UDP需要配置两个端口：一个用于广播接收，一个用于服务器通信
 * \warning 必须确保网络地址有效且端口未被占用
 */
bool ParserUDP::init( WTSVariant* config )
{
	_hots = config->getCString("host");
	_bport = config->getInt32("bport");
	_sport = config->getInt32("sport");
	_gpsize = config->getUInt32("gpsize");
	if (_gpsize == 0)
		_gpsize = 1000;

			ip::address addr = ip::make_address(_hots);
	_server_ep = ip::udp::endpoint(addr, _sport);

	_broad_ep = ip::udp::endpoint(ip::address_v4::any(), _bport);

	return true;
}

/*!
 * \brief 释放资源
 * 
 * \details 释放UDP行情解析器占用的所有资源，当前实现为空
 * 
 * \note 资源清理主要通过disconnect()实现
 */
void ParserUDP::release()
{
	
}

/*!
 * \brief 重新连接UDP服务器
 * \return 是否重连成功
 * 
 * \details 重新建立UDP连接，包括两个套接字的创建和配置：
 *          
 *          广播通道设置：
 *          - 关闭并重建广播套接字
 *          - 设置地址重用选项
 *          - 启用广播功能
 *          - 设置8MB接收缓冲区
 *          - 绑定到广播端口
 *          - 启动异步接收
 *          
 *          服务器通道设置：
 *          - 关闭并重建服务器套接字
 *          - 创建UDP v4套接字
 *          - 启动异步接收
 *          - 发送订阅请求
 * 
 * \note 使用boost::bind绑定异步回调函数
 * \warning 套接字创建失败会导致连接异常
 */
bool ParserUDP::reconnect()
{
	{// 设置广播通道
		if (_b_socket != NULL)
		{
			_b_socket->close();
			delete _b_socket;
			_b_socket = NULL;
		}

		_b_socket = new ip::udp::socket(_io_service);

		_b_socket->open(_broad_ep.protocol());
		_b_socket->set_option(ip::udp::socket::reuse_address(true));
		_b_socket->set_option(ip::udp::socket::broadcast(true));
		_b_socket->set_option(ip::udp::socket::receive_buffer_size(8 * 1024 * 1024));
		_b_socket->bind(_broad_ep);


		_b_socket->async_receive_from(buffer(_b_buffer), _broad_ep,
			boost::bind(&ParserUDP::handle_read, this,
			boost::asio::placeholders::error,
			boost::asio::placeholders::bytes_transferred, true));
	}

	{
		// 设置服务器通道
		if (_s_socket != NULL)
		{
			_s_socket->close();
			delete _s_socket;
			_s_socket = NULL;
		}

		_s_socket = new ip::udp::socket(_io_service, ip::udp::endpoint(ip::udp::v4(), 0));

		_s_socket->async_receive_from(buffer(_s_buffer), _server_ep,
			boost::bind(&ParserUDP::handle_read, this,
			boost::asio::placeholders::error,
			boost::asio::placeholders::bytes_transferred, false));

		subscribe();
	}
	return true;
}

/*!
 * \brief 发送订阅请求
 * 
 * \details 向UDP服务器发送合约订阅请求：
 *          - 创建订阅请求数据包
 *          - 遍历所有待订阅的合约代码
 *          - 处理交易所前缀（去除"交易所."前缀）
 *          - 使用逗号分隔多个合约代码
 *          - 支持数据包分片（单包最大1000字节）
 *          - 将请求加入发送队列
 *          - 启动异步发送流程
 * 
 * \note 当单个数据包长度超过1000字节时会自动分片发送
 * \warning 必须确保合约代码格式正确
 */
void ParserUDP::subscribe()
{
	std::string data;
	data.resize(sizeof(UDPReqPacket), 0);
	UDPReqPacket* req = (UDPReqPacket*)data.data();
	req->_type = UDP_MSG_SUBSCRIBE;
	uint32_t length = 0;
	for (auto& code : _set_subs)
	{
		if (length > 0)
		{
			req->_data[length] = ',';
			length++;
		}

		std::size_t pos = code.find('.');
		if (pos != std::string::npos)
			strcpy(req->_data + length, (char*)code.c_str() + pos + 1);
		else
			strcpy(req->_data + length, code.c_str());

		length += code.size();

		if (length > 1000)
		{
			_send_queue.push(data);
			
			data.resize(sizeof(UDPReqPacket), 0);
			req = (UDPReqPacket*)data.data();
			req->_type = UDP_MSG_SUBSCRIBE;
			length = 0;
		}
	}

	do_send();

	write_log(_sink, LL_INFO, "[ParserUDP] Ticks subscribing sent");
}

/*!
 * \brief 执行数据发送
 * 
 * \details 从发送队列中取出数据进行异步发送：
 *          - 检查发送队列是否为空
 *          - 获取队列头部的数据
 *          - 使用异步发送到服务器终端点
 *          - 绑定发送完成回调函数
 * 
 * \note 发送完成后会在handle_write中处理队列清理
 */
void ParserUDP::do_send()
{
	if (_send_queue.empty())
		return;

	std::string& data = _send_queue.front();

	_s_socket->async_send_to(boost::asio::buffer(data, data.size()), _server_ep,
		boost::bind(&ParserUDP::handle_write, this, boost::asio::placeholders::error));
}

/*!
 * \brief 处理UDP数据发送完成
 * \param e boost错误代码
 * 
 * \details 处理异步发送完成事件：
 *          - 检查发送是否成功
 *          - 记录发送错误信息
 *          - 从发送队列中移除已发送的数据
 *          - 继续发送队列中的下一个数据
 * 
 * \note 即使发送失败也会移除队列头部数据，继续后续发送
 */
void ParserUDP::handle_write(const boost::system::error_code& e)
{
	if (e)
	{
		write_log(_sink, LL_ERROR, "[ParserUDP] Error occured while receiving: {}({})", e.message().c_str(), e.value());
	}
	else
	{
		_send_queue.pop();
	}
	
	do_send();
}

/*!
 * \brief 建立连接
 * \return 是否连接成功
 * 
 * \details 启动UDP网络连接：
 *          - 调用reconnect()建立UDP套接字连接
 *          - 创建网络处理线程运行io_service
 *          - 开始异步网络事件循环
 * 
 * \note 网络处理在独立线程中运行，避免阻塞主线程
 */
bool ParserUDP::connect()
{
	if(reconnect())
	{
		_thrd_parser.reset(new StdThread(boost::bind(&io_service::run, &_io_service)));
	}
	else
	{
		return false;
	}

	return true;
}

/*!
 * \brief 断开连接
 * \return 是否断开成功
 * 
 * \details 安全断开UDP连接：
 *          - 关闭并删除广播套接字
 *          - 设置停止标志
 *          - 通过strand投递断开事件处理
 *          - 确保线程安全的断开流程
 * 
 * \note 使用strand确保断开操作的线程安全性
 */
bool ParserUDP::disconnect()
{
	if(_b_socket != NULL)
	{
		_b_socket->close();
		delete _b_socket;
		_b_socket = NULL;
	}

	_stopped = true;
	boost::asio::post(_strand, boost::bind(&ParserUDP::doOnDisconnected, this));

	return true;
}

/*!
 * \brief 检查连接状态
 * \return 是否已连接
 * 
 * \details 通过检查广播套接字是否有效来判断连接状态
 * 
 * \note 以广播套接字为准判断连接状态
 */
bool ParserUDP::isConnected()
{
	return _b_socket!=NULL;
}

/*!
 * \brief 订阅行情数据
 * \param vecSymbols 要订阅的合约代码集合
 * 
 * \details 将合约代码添加到订阅集合：
 *          - 遍历传入的合约代码
 *          - 检查代码是否已存在于订阅集合中
 *          - 避免重复订阅同一合约
 *          - 将新代码加入订阅集合
 * 
 * \note 订阅集合用于在连接时发送订阅请求
 */
void ParserUDP::subscribe( const CodeSet &vecSymbols )
{
	auto cit = vecSymbols.begin();
	for(; cit != vecSymbols.end(); cit++)
	{
		const auto &code = *cit;
		if(_set_subs.find(code) == _set_subs.end())
		{
			_set_subs.insert(code);
		}
	}
}

/*!
 * \brief 退订行情数据
 * \param setSymbols 要退订的合约代码集合
 * 
 * \details 处理行情退订请求，当前实现为空，可根据需要扩展
 * 
 * \note 预留接口，用于实现行情数据的动态退订功能
 */
void ParserUDP::unsubscribe(const CodeSet &setSymbols)
{

}

/*!
 * \brief 注册行情回调接口
 * \param listener 行情数据回调接口指针
 * 
 * \details 注册行情数据处理接口：
 *          - 检查是否替换了已有的监听器
 *          - 保存新的回调接口指针
 *          - 记录监听器替换的警告信息
 * 
 * \note 如果替换已有监听器会记录警告日志
 */
void ParserUDP::registerSpi( IParserSpi* listener )
{
	bool bReplaced = (_sink!=NULL);
	_sink = listener;
	if(bReplaced && _sink)
	{
		write_log(_sink, LL_WARN, "Listener is replaced");
	}
}

/*!
 * \brief 处理UDP数据接收
 * \param e boost错误代码
 * \param bytes_transferred 接收到的字节数
 * \param isBroad 是否为广播通道数据
 * 
 * \details 异步处理UDP数据接收：
 *          - 检查接收是否发生错误
 *          - 错误时触发断开事件和重连机制
 *          - 成功时调用extract_buffer解析数据
 *          - 继续启动下一次异步接收
 *          - 分别处理广播通道和服务器通道
 * 
 * \note 网络错误时会自动尝试重连（2秒延迟）
 * \warning 停止状态下不会进行重连和数据处理
 */
void ParserUDP::handle_read(const boost::system::error_code& e, std::size_t bytes_transferred, bool isBroad /* = true */)
{
	if(e)
	{
		if(_sink)
			_sink->handleEvent(WPE_Close, 0);

		write_log(_sink, LL_ERROR, "[ParserUDP] Error occured while receiving: {}({})", e.message().c_str(), e.value());

		if (!_stopped && !_connecting)
		{
			std::this_thread::sleep_for(std::chrono::seconds(2));
			reconnect();
			return;
		}
	}

	if(_stopped || bytes_transferred<=0)
		return;

	extract_buffer(bytes_transferred, isBroad);

	if (isBroad && _b_socket)
	{
		_b_socket->async_receive_from(buffer(_b_buffer), _broad_ep,
			boost::bind(&ParserUDP::handle_read, this, 
			boost::asio::placeholders::error,
			boost::asio::placeholders::bytes_transferred, true));
	} 
	else if(!isBroad && _s_socket)
	{
		_s_socket->async_receive_from(buffer(_s_buffer), _server_ep,
			boost::bind(&ParserUDP::handle_read, this,
			boost::asio::placeholders::error,
			boost::asio::placeholders::bytes_transferred, false));

	}
}

/*!
 * \brief 解析UDP数据缓冲区
 * \param length 数据长度
 * \param isBroad 是否为广播数据
 * 
 * \details 解析接收到的UDP数据包，根据消息类型分发处理：
 *          
 *          支持的数据类型：
 *          - UDP_MSG_PUSHTICK/UDP_MSG_SUBSCRIBE: Tick行情数据
 *          - UDP_MSG_PUSHORDDTL: 委托明细数据
 *          - UDP_MSG_PUSHORDQUE: 委托队列数据
 *          - UDP_MSG_PUSHTRANS: 逐笔成交数据
 *          
 *          处理流程：
 *          - 解析数据包头获取消息类型
 *          - 根据类型创建对应的WonderTrader数据对象
 *          - 调用相应的回调函数分发数据
 *          - 记录接收统计信息（按gpsize间隔输出）
 * 
 * \note 每种数据类型都有独立的接收计数统计
 * \warning 未知的消息类型会被忽略
 */
void ParserUDP::extract_buffer(uint32_t length, bool isBroad /* = true */)
{
	UDPPacketHead* header = NULL;
	if(isBroad)
		header = (UDPTickPacket*)_b_buffer.data();
	else
		header = (UDPTickPacket*)_s_buffer.data();

	if (header->_type == UDP_MSG_PUSHTICK || header->_type == UDP_MSG_SUBSCRIBE)
	{
		UDPTickPacket* packet = (UDPTickPacket*)header;
		WTSTickData* curTick = WTSTickData::create(packet->_data);
		if (_sink)
			_sink->handleQuote(curTick, 0);

		curTick->release();

		static uint32_t recv_cnt = 0;
		recv_cnt++;
		if (recv_cnt % _gpsize == 0)
			write_log(_sink, LL_DEBUG, "[ParserUDP] {} ticks received in total", recv_cnt);
	}
	else if (header->_type == UDP_MSG_PUSHORDDTL)
	{
		UDPOrdDtlPacket* packet = (UDPOrdDtlPacket*)header;
		WTSOrdDtlData* curData = WTSOrdDtlData::create(packet->_data);
		if (_sink)
			_sink->handleOrderDetail(curData);

		curData->release();

		static uint32_t recv_cnt = 0;
		recv_cnt++;
		if (recv_cnt % _gpsize == 0)
			write_log(_sink, LL_DEBUG, "[ParserUDP] {} order details received in total", recv_cnt);
	}
	else if (header->_type == UDP_MSG_PUSHORDQUE)
	{
		UDPOrdQuePacket* packet = (UDPOrdQuePacket*)header;
		WTSOrdQueData* curData = WTSOrdQueData::create(packet->_data);
		if (_sink)
			_sink->handleOrderQueue(curData);

		curData->release();

		static uint32_t recv_cnt = 0;
		recv_cnt++;
		if (recv_cnt % _gpsize == 0)
			write_log(_sink, LL_DEBUG, "[ParserUDP] {} order queues received in total", recv_cnt);
	}
	else if (header->_type == UDP_MSG_PUSHTRANS)
	{
		UDPTransPacket* packet = (UDPTransPacket*)header;
		WTSTransData* curData = WTSTransData::create(packet->_data);
		if (_sink)
			_sink->handleTransaction(curData);

		curData->release();

		static uint32_t recv_cnt = 0;
		recv_cnt++;
		if (recv_cnt % _gpsize == 0)
			write_log(_sink, LL_DEBUG, "[ParserUDP] {} transactions received in total", recv_cnt);
	}
}

/*!
 * \brief 处理连接成功事件
 * 
 * \details 连接成功后的回调处理：
 *          - 触发WPE_Connect连接事件
 *          - 触发WPE_Login登录事件
 * 
 * \note UDP是无连接协议，此方法主要用于状态通知
 */
void ParserUDP::doOnConnected()
{
	if(_sink)
	{
		_sink->handleEvent(WPE_Connect, 0);
		_sink->handleEvent(WPE_Login, 0);
	}
}

/*!
 * \brief 处理连接断开事件
 * 
 * \details 连接断开后的回调处理：
 *          - 触发WPE_Close断开事件
 * 
 * \note 通知上层应用连接已断开，可以进行相应的清理工作
 */
void ParserUDP::doOnDisconnected()
{
	if(_sink)
	{
		_sink->handleEvent(WPE_Close, 0);
	}
}